#include "sql_writer.h"
#include <sstream>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include <algorithm>
#include <cstdio>
#include <cstdlib>
#include <ctime>
#include <iomanip>
#include <iostream>
#include <chrono>

using kwdb::ingest::v1::Point;
using kwdb::ingest::v1::WriteRequest;

// Helper: safe escaping for SQL identifiers
static std::string quoteIdent(const std::string& s) {
  std::string out = "\"";
  for (char c : s) out += (c == '"' ? "\"\"" : std::string(1, c));
  out += "\""; return out;
}

static std::string quoteLiteral(const std::string& s) {
  std::string out = "'";
  for (char c : s) out += (c == '\'' ? "''" : std::string(1, c));
  out += "'"; return out;
}

static std::string ts_ns_to_sql(int64_t ns) {
  // Convert nanoseconds to SQL timestamp string
  time_t sec = ns / 1000000000LL;
  int64_t nsec = ns % 1000000000LL;
  std::tm tm; gmtime_r(&sec, &tm);
  std::ostringstream oss;
  oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S") << "." << std::setw(9) << std::setfill('0') << nsec;
  return quoteLiteral(oss.str());
}

// Group points by measurement
static std::unordered_map<std::string, std::vector<const Point*>> group_by_meas(const WriteRequest& req) {
  std::unordered_map<std::string, std::vector<const Point*>> g;
  for (const auto& p : req.points()) g[p.measurement()].push_back(&p);
  return g;
}

// Execute SQL using psql command
static long long exec_sql(const std::string& sql, std::string* err) {
  const char* host = std::getenv("KWDB_HOST");
  const char* port = std::getenv("KWDB_SQL_PORT");
  const char* db   = std::getenv("KWDB_DB");
  const char* user = std::getenv("KWDB_USER");
  const char* pwd  = std::getenv("KWDB_PWD");
  const char* kwbase_path = std::getenv("KWBASE_PATH");

  std::ostringstream cmd;

  // Debug: Print environment variables
  std::cout << "[exec_sql] KWBASE_PATH=" << (kwbase_path ? kwbase_path : "NULL") << std::endl;

  // Use kwbase sql if KWBASE_PATH is set
  if (kwbase_path && *kwbase_path) {
    // KWDB native client
    std::cout << "[exec_sql] Using kwbase sql client" << std::endl;
    cmd << "cd " << kwbase_path << " && "
        << "export LD_LIBRARY_PATH=../lib && "
        << "./kwbase sql --insecure"
        << " --host=" << (host?host:"127.0.0.1") << ":" << (port?port:"26257")
        << " --database=" << (db?db:"kwdb")
        << " -e \"" << sql << "\" 2>&1";
  } else {
    // PostgreSQL client (psql)
    std::cout << "[exec_sql] Using psql client" << std::endl;
    if (pwd && *pwd) cmd << "PGPASSWORD='" << pwd << "' ";
    cmd << "psql"
        << " -h " << (host?host:"127.0.0.1")
        << " -p " << (port?port:"5432")
        << " -U " << (user?user:"kwdb")
        << " -d " << (db?db:"kwdb")
        << " -t -c \"" << sql << "\" 2>&1";
  }

  std::cout << "[exec_sql] CMD: " << cmd.str() << std::endl;

  FILE* pipe = popen(cmd.str().c_str(), "r");
  if (!pipe) { if (err) *err = "popen failed"; return -1; }
  std::string out; char buf[4096];
  while (fgets(buf, sizeof(buf), pipe)) out += buf;
  int rc = pclose(pipe);

  if (rc == 0) return 1;
  if (err) *err = out;
  return -1;
}

long long write_points_sql(const WriteRequest& req, std::string* err) {
  auto start = std::chrono::steady_clock::now();
  int total_points = req.points_size();

  std::cout << "[sql_writer] recv " << total_points << " points" << std::endl;

  auto groups = group_by_meas(req);
  long long total = 0;

  for (auto& kv : groups) {
    const std::string& meas = kv.first;
    const auto& vec = kv.second;

    // Validate measurement name
    if (meas.empty()) {
      if (err) *err = "empty measurement name";
      return -1;
    }

    // Collect column names: ts + all tags/fields (union of all points)
    std::unordered_set<std::string> tag_set;
    std::unordered_set<std::string> field_set;

    for (const auto* p : vec) {
      for (const auto& t : p->tags())   tag_set.insert(t.key());
      for (const auto& f : p->fields()) field_set.insert(f.key());
    }

    // Convert to vectors for stable ordering
    std::vector<std::string> tag_cols(tag_set.begin(), tag_set.end());
    std::vector<std::string> field_cols(field_set.begin(), field_set.end());
    std::sort(tag_cols.begin(), tag_cols.end());
    std::sort(field_cols.begin(), field_cols.end());

    // Build INSERT statement
    std::ostringstream sql;
    sql << "INSERT INTO " << quoteIdent(meas) << "("
        << quoteIdent("ts");
    for (auto& c : tag_cols)   sql << "," << quoteIdent(c);
    for (auto& c : field_cols) sql << "," << quoteIdent(c);
    sql << ") VALUES ";

    // Build VALUES list
    bool first_row = true;
    for (const auto* p : vec) {
      if (!first_row) sql << ",";
      first_row = false;

      // Build tag/field maps for lookup
      std::unordered_map<std::string, std::string> tags;
      for (const auto& t : p->tags()) tags[t.key()] = t.value();
      std::unordered_map<std::string, double> fields;
      for (const auto& f : p->fields()) fields[f.key()] = f.f64();

      sql << "(" << ts_ns_to_sql(p->timestamp_unix_ns());
      for (auto& c : tag_cols) {
        auto it = tags.find(c);
        sql << "," << (it==tags.end() ? "NULL" : quoteLiteral(it->second));
      }
      for (auto& c : field_cols) {
        auto it = fields.find(c);
        if (it==fields.end()) sql << ",NULL";
        else                  sql << "," << it->second;
      }
      sql << ")";
    }
    sql << ";";

    long long n = exec_sql(sql.str(), err);
    if (n < 0) return -1;
    total += static_cast<long long>(vec.size());
  }

  auto end = std::chrono::steady_clock::now();
  auto duration_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();

  std::cout << "[sql_writer] written=" << total << " points, elapsed=" << duration_ms << "ms" << std::endl;

  return total;
}